#ifndef _KAFKA_CONSUMER_H_H_H
#define _KAFKA_CONSUMER_H_H_H

#include "kafka_eventcb.hpp"

#include <librdkafka/rdkafkacpp.h>
#include <memory>

class KafkaConsumer
{
public:
    KafkaConsumer(const std::string &brokers, const std::string &groupId);
    ~KafkaConsumer();

public:
    bool                              init(bool               with_sasl   = false,
                                           const std::string &protocol    = "",
                                           const std::string &mechanisms  = "",
                                           const std::string &username    = "",
                                           const std::string &password    = "",
                                           bool               with_ssl    = false,
                                           const std::string &ca_location = "");
    bool                              subscribe(const std::string &topic_name, int32_t partition);
    bool                              subscribe(const std::string &topic_name, int32_t partition, const std::string &offset);
    bool                              unsubscribe();
    std::shared_ptr<RdKafka::Message> consume(int timeout_ms);

private:
    RdKafka::Consumer *m_consumer;
    RdKafka::Topic    *m_topic;
    RdKafkaEventCb    *m_eventcb;
    int32_t            m_partition;
    std::string        m_brokers;
    std::string        m_groupId;
};

#endif